package hy;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.IOException;
import java.util.Arrays;
import java.util.Properties;

public class MyConsumer {
    public static void main(String[] args) throws IOException {
        Properties p=new Properties();
        p.load(MyConsumer.class.getClassLoader().getResourceAsStream("consumer.properties"));

        KafkaConsumer<String,String> consumer=new KafkaConsumer<String, String>(p);
        int i=0;
        while(true){
            consumer.subscribe(Arrays.asList("first","second"));
            ConsumerRecords<String,String> cr=consumer.poll(100);

            for(ConsumerRecord<String,String> r:cr){
                System.out.println("主题：" + r.topic() + "，所属分区：" + r.partition() + ",offset:" + r.offset() + ",内容：" + r.value());
                //consumer.commitSync();
                if(++i==2){
                    //throw new NullPointerException();
                }
            }
        }
        //手动提交
        /*
        * 异步：commitAsync,提交后就结束，不阻塞
        * 同步：commitSync,提交后会等待kafka确认，在收到确认前会阻塞
        * */
    }
}